import asyncio
import copy
import datetime
import json
import traceback

from apps.database import get_ods, mongo, save_ods, save_ods_result, get_ods_result_by_id, save_dwd, update_ods
from apps.global_config import configs
from apps.models.BaseModel import OdsResultModel, UpdateDwdModel
from apps.models.QueryModel import QueryDetail, QuerySchema
from apps.models.ResponseModel import err_info
from apps.task.parse_calendar import ParseCalendar
from apps.utils.MinioBoto3 import S3Util


async def get_task():
    qd = QueryDetail(field_name='source_type', field_value=2, link='and', oparation='eq')
    qd1 = QueryDetail(field_name='status', field_value=0, link='and', oparation='eq')
    query = QuerySchema(
        query=[qd, qd1],
        limit=100
    )
    results = await get_ods(query.get_filter(),
                            query.get_projection(),
                            query.get_sort_by(),
                            query.skip,
                            query.limit)
    for result in results:
        yield result


async def run():
    # mongo.init(configs.MONGO_URI)
    async for item in get_task():
        print(item["lngid"])
        try:
            bools, results = await parse_task(item)
            if not bools:
                item["status"] = 101
                item["error_info"] = json.dumps(results, ensure_ascii=False)
                await save_ods({"_id": item["_id"]}, item)
            else:
                ods_result = OdsResultModel(lngid=results["lngid"],
                                            rawid=results["rawid"],
                                            process_id=results["process_id"],
                                            source_type=results["source_type"],
                                            type_id=results["type_id"],
                                            latest_date=results["latest_date"],
                                            result=results)
                save_result = await save_ods_result(ods_result.lngid,
                                                    ods_result.lngid + '.json',
                                                    ods_result.get_result(),
                                                    ods_result.get_metadata())
                if save_result["code"] == 200:
                    item["status"] = 1
                    item["error_info"] = ''
                    await save_ods({"_id": item["_id"]}, item)
                else:
                    item["status"] = save_result["code"]
                    item["error_info"] = save_result['msg']
                    await save_ods({"_id": item["_id"]}, item)
        except:
            traceback.print_exc()
            raise Exception("save_ods 保存状态错误")


async def read_file(item):
    all_file_contents = ""
    list_count = []
    lists = ["processtxt", str(item["source_type"]), item['process_id'], item['lngid']]
    file_lists = sorted(item["file_name"])
    for file_name in file_lists:
        tp_list = copy.deepcopy(lists)
        tp_list.append(file_name)
        key = '/'.join(tp_list)
        s3_result = await S3Util.read_s3_data(key)
        if s3_result["code"] != 200:
            return s3_result, '', False
        all_file_contents = all_file_contents + s3_result["data"]
        list_count.append((len(s3_result["data"].split("\n")) - 1, key))

    all_count = sum(i for i, v in list_count)
    if len(file_lists) > 1 and all_count + 1 != len(all_file_contents.split('\n')):
        raise Exception("合并文件长度问题")
    return all_file_contents, list_count, True


async def parse_task(item):
    pc = ParseCalendar()
    all_file_contents, list_count, bools = await read_file(item)
    if not bools:
        return False, all_file_contents
    pc.set_inputs(all_file_contents)
    pc.set_file_info(list_count)
    try:
        pc.data_parse(item)
        result = pc.get_parse_result(item)
        return True, result
    except Exception as e:
        from apps.task.works_one_parse import WorksOneParse
        dicts = err_info()
        dicts["err_msg"] = repr(e)
        dicts["处理到行"] = str(pc.cop.now_deal_index_line)
        dicts["已处理长度"] = str(pc._finish_deal_len)
        if pc.cop.now_deal_index_line:
            if index_line := pc.cop.now_deal_index_line[-1]:
                file_in_count = index_line[0] + 1  # +1 是因为 行与 index 之间相差1
                index, filename = pc.get_count_in_file(file_in_count)
                dicts["处理位置"] = f"处理到行:{index}, 在文件{filename}"
        dicts["traceback"] = traceback.format_exc()
        pc.cop.now_deal_index_line.clear()
        return False, dicts


async def run_dwd():
    # mongo.init(configs.MONGO_URI)
    qd = QueryDetail(field_name='status', field_value=1, link='and', oparation='eq')
    qd1 = QueryDetail(field_name='source_type', field_value=2, link='and', oparation='eq')
    query = QuerySchema(
        query=[qd, qd1],
        projection=['lngid']
    )

    result = await get_ods(query.get_filter(),
                           query.get_projection(),
                           query.get_sort_by(),
                           query.skip,
                           query.limit)

    for one_obs in result:
        try:
            result_dicts = await get_ods_result_by_id(one_obs["lngid"])
            cclp_ods = result_dicts["data"]
            dwd_date = UpdateDwdModel(
                m_id=cclp_ods["rawid"] + "_" + cclp_ods["latest_date"],
                ods_id=cclp_ods["_id"],
                user_id=0,
                keyid=cclp_ods["rawid"],
                rawid=cclp_ods["rawid"],
                process_id=cclp_ods["process_id"],
                source_type=cclp_ods["source_type"],
                type_id=cclp_ods["type_id"],
                is_deprecated=0,
                latest_date=cclp_ods["latest_date"],
                update_date=datetime.datetime.now().strftime("%Y%m%d%H%M%S%f"),
                date_info=cclp_ods["date_info"]
            )
            dwd_result = await save_dwd({'_id': dwd_date.m_id}, dwd_date.get_dict())
            if dwd_result["code"] != 200:
                dicts = err_info()
                dicts["err_msg"] = dwd_result["msg"]
                update_dict = {
                    'status': dwd_result["code"],
                    "error_info": json.dumps(dicts, ensure_ascii=False)
                }
                await update_ods({"_id": one_obs["_id"]}, update_dict)
            else:
                update_dict = {
                    'status': 2
                }
                await update_ods({"_id": one_obs["lngid"]}, update_dict)
        except Exception as e:
            update_dict = {}
            update_dict["status"] = 101
            dicts = err_info()
            dicts["traceback"] = traceback.format_exc()
            dicts["err_msg"] = repr(e)
            update_dict["error_info"] = json.dumps(dicts, ensure_ascii=False)
            await update_ods({"_id": one_obs["_id"]}, update_dict)


async def run_task():
    await run()
    await run_dwd()


if __name__ == "__main__":
    asyncio.run(run_task())
